package cn.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * 测试消费逻辑
 */
@Slf4j
public class MyConsumer {
    @Test
    public void consume01() throws Exception {
        //1. 连接rocketmq的消费者对象
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer();
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setConsumerGroup("consumer-group-02");
        //提供消费者监听的目标信息 监听哪个主题 test-topic-01 * 表示监听条件,是主题中所有消息
        consumer.subscribe("test-topic-03","*");
        //2. 建立消费的监听 消费逻辑(打印)
        MessageListenerConcurrently messageListener=new MessageListenerConcurrently() {
            /**每次 rocketmq推送消息,都会让监听器执行这个consumeMessage方法
             * 消费逻辑的 核心方法
             * @param list MessageExt 推送过来的消息 当前不是批量消费,所以每个消息都会在
             *             list中存在唯一一个元素
             * @param consumeConcurrentlyContext 消费上下文 拿到消费者 id 等信息
             * @return 反馈给rocketmq的消费状态 CONSUME_SUCCESS RECONSUME_LATER(重试retry机制)
             */
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //从list元素中 解析需要的数据 body
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();
                String msg=new String(body, StandardCharsets.UTF_8);
                log.info("消费者1接收到消息:{}",msg);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        };
        //创建的监听器,交给消费者使用
        consumer.setMessageListener(messageListener);
        //开启consumer连接
        consumer.start();
        //3. TODO 可以关闭
        //4. 保证当前测试线程 死循环
        while(true);
    }
}
